package com.webull.information.center.carwler.common.util;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * @author smj
 * @version jdk1.8
 * @Description
 * @Date 下午7:12:10
 */
public class NewsTaskUtil {
    protected static final Logger logger = LogManager.getLogger(NewsTaskUtil.class);


	public static void main(String[] args) {
		System.out.println(ArrayUtils.toString(Arrays.asList(1, 2, 4)));
		System.out.println(ArrayUtils.toString(1));

		// List<Integer> a = Arrays.asList(1, 2, 3, 4, 5, 123, 66, 77, 8, 8, 99, 9, 1,
		// 54, 14, 7, 6, 23, 5, 44, 54, 3, 55,
		// 3, 43, 534, 5, 345, 46, 45645);
		// ThreadPoolTaskExecutor ttt = new ThreadPoolTaskExecutor();
		// ttt.initialize();
		// Consumer<Integer> dd = cc -> {
		// try {
		// Thread.sleep(500l);
		// } catch (InterruptedException e) {
		// }
		// System.out.println(cc);
		// };
		// NewsTaskUtil.multiThreadProcess(a, dd, ttt);
		// System.out.println("&&&&&&&&");
		// ttt.destroy();
	}

	public static <T> int multiThreadProcess(List<T> list, Consumer<T> process,
                                              ThreadPoolTaskExecutor threadPoolTaskExecutor, int latch) {

		int errors = 0;
		if (list == null || list.isEmpty())
			return errors;
        if (latch <= 1) throw new IllegalArgumentException("latch must > 1");
        /*
         * 切分处理，每次处理最多latch条任务
		 */

        final T[] waits = (T[]) new Object[latch];
		int latW = 0;// 批次下任务数量
		final int length0 = list.size() - 1;
		for (int i = 0; i <= length0; i++) {
            waits[latW++] = list.get(i);
			if (latW >= latch || i >= length0) { // 达到一个批次的执行条件
				errors += processLatch(latW, waits, process, threadPoolTaskExecutor);

                for (int j = 0; j < waits.length; j++) {// 清空临时数组
                    waits[j] = null;
                }
                latW = 0;
            }
        }
		return errors;
    }

    /**
     * 多线程处理新闻等任务
     *
     * @param list                   处理的任务列表
     * @param process                单条任务处理逻辑
     * @param threadPoolTaskExecutor 线程池
     */
	public static <T> int multiThreadProcess(List<T> list, Consumer<T> process,
                                              ThreadPoolTaskExecutor threadPoolTaskExecutor) {
		return multiThreadProcess(list, process, threadPoolTaskExecutor, 10);
    }

	private static <T> int processLatch(int latch, final T[] waits, Consumer<T> process,
			ThreadPoolTaskExecutor threadPoolTaskExecutor) { // 执行批次，必须该批次执行完后，再执行下个批次
		int[] errors = new int[] { 0 };
		try {
            final CountDownLatch latchs = new CountDownLatch(latch);
            for (int i = 0; i < latch; i++) {
                final T t = waits[i];
                threadPoolTaskExecutor.execute(() -> {
                    try {
                        if (t != null)
                            process.accept(t);
                    } catch (Exception e) {
						errors[0]++;
						logger.info("TaskUtil.procss t:{},p:{} latch:Exception:", ArrayUtils.toString(t),
								process.getClass().getName(), e);
                    } finally {
                        latchs.countDown();
                    }
                });
            }
            logger.info("TaskUtil.procss latch:{}", latch);
            latchs.await();
        } catch (Exception e) {
			errors[0]++;
            logger.info("TaskUtil.procss CountDownLatch error:{}", e);
        }
		return errors[0];

    }
}
